
----------------------------------------------------------------
--- Copyright：  迷你玩-海马
--- Description：各业务服务器连接到data_cache缓存服务器时使用，
--- 使用ngx自带的连接池，长连接到data_cache并发送读写请求，
--- 业务服务器 --(mixDcClient.lua)--> data_cache缓存服务器 ----> (SSDB数据库/mysql数据库)
---
---
--- mix ( 区别于其他xxxDcClient,  mix会进行双写和双读，目前用来做ssdb数据到mysql数据的转换 )
--- 支持下面3中模式：
---		1 仅读写ssdb落地       (切换前)
---		2 同时读写ssdb和mysql  (切换期间) (使用scan_sync同步数据)
---		3 仅读写mysql落地      (切换后)
---
---
--- ssdb 向 mysql 数据库迁移解耦过渡期使用（ 同步完成后一直使用也没问题 ）
---
----------------------------------------------------------------

local assert     = assert
local require    = require
local bit        = bit
local coroutine  = coroutine
local debug      = debug
local io         = io
local pairs      = pairs
local ipairs     = ipairs
local math       = math
local os         = os
local print      = print
local pcall      = pcall
local xpcall     = xpcall
local rawget     = rawget
local rawset     = rawset
local select     = select
local string     = string
local table      = table
local tonumber   = tonumber
local tostring	 = tostring
local error      = error
local type       = type
local unpack     = unpack
local setmetatable = setmetatable
local getmetatable = getmetatable
local ngx = ngx

require "resty.core.shdict"

local ns_env     = require("modules.env.env")
local log        = require("modules.commhm.log");
local ns_time    = require("modules.commhm.time");
local uu         = require("modules.commhm.utils");
local ns_network = require("modules.commhm.network");

local ns_ssdbDcClient  = require("modules.commhm.dc.ssdbDcClient");
local ns_mysqlDcClient = require("modules.commhm.dc.mysqlDcClient");


-- SSDB MGR类
local _M = { _VERSION = '0.12.1' }
local mt = { __index = _M }


local  __MIX_ORDER_SSDB_MYSQL_0__ = 0    --0=先ssdb后mysql
local  __MIX_ORDER_MYSQL_SSDB_1__ = 1    --1=先mysql后ssdb


---构造函数，使用配置文件初始化两个data_cache连接  ns_mixDcClient
function _M.new(self, module_name_,
				ssdb_config_, mysql_config_,
				order_type_, params_table_ )

	log.debug( log.fast() and "call mixDcClient:new" );
	local ins_ = {
		config_cache = {};
		module_name   = module_name_;

		ssdbDcClient  = nil,          -- ssdb落地
		mysqlDcClient = nil,          -- mysql落地
		order_type    = nil,          -- 顺序类型 (nil=单种落地  0=先ssdb后mysql  1=先mysql后ssdb )

		if_check_read_ret  = false,   -- 是否比较两边read的结果                1=比较  其他=不同步
		if_check_mread_ret = false,   -- 是否比较两边mread的结果               1=比较  其他=不同步
		repair_data        = false,   -- 当1和2数据不一致的时候，是否进行修复     1=修复  其他=不修复
	};

	-- ssdb
	if  ssdb_config_ and ssdb_config_.server and ssdb_config_.select then
		ins_.ssdbDcClient  = ns_ssdbDcClient:new( module_name_,
				ssdb_config_.server, ssdb_config_.select )
	end

	-- mysql
	if  mysql_config_ and mysql_config_.server and mysql_config_.select then
		ins_.mysqlDcClient = ns_mysqlDcClient:new( module_name_,
				mysql_config_.server, mysql_config_.select )
	end


	-- 两种落地方式都有的情况（ssdb+mysql）
	if  ins_.ssdbDcClient and ins_.mysqlDcClient then
		ins_.order_type = (order_type_ or __MIX_ORDER_SSDB_MYSQL_0__)         -- 选择优先落地方式

		-- 控制参数
		if  params_table_ then
			if  params_table_.if_check_read_ret then
				ins_.if_check_read_ret  = params_table_.if_check_read_ret     -- 是否比较两边read的结果
			end

			if  params_table_.if_check_mread_ret then
				ins_.if_check_mread_ret = params_table_.if_check_mread_ret    -- 是否比较两边mread的结果
			end

			if  params_table_.repair_data then
				ins_.repair_data = params_table_.repair_data                  -- 是否修复不同的数据
			end
		end
	end

    return setmetatable(ins_, mt);
end


---释放当次请求中的所有远程锁， 一般在 gg.final() 中调用
---扫描当前请求中，哪些地方使用了远程锁，并全部释放掉
function _M.final( self )

	local function func_ssdb_()
		if  self.ssdbDcClient then
			self.ssdbDcClient:final()
		end
	end
	local function func_mysql_()
		if  self.mysqlDcClient then
			self.mysqlDcClient:final()
		end
	end

	if  self.order_type then
		local thread_list_ = {}
		uu.thread_spawn( func_ssdb_,  thread_list_ )
		uu.thread_spawn( func_mysql_, thread_list_ )
		uu.thread_wait( thread_list_ )
	else
		func_ssdb_()
		func_mysql_()
	end

end


---解锁全部远程锁，请求过程中途主动释放，很少有地方使用
function _M.unlock_all( self )
	self:final()
end


---解锁指定kh远程锁
function _M.unlock_key_hash( self, select_, key_, hash_ )

	local function func_ssdb_()
		if  self.ssdbDcClient then
			self.ssdbDcClient:unlock_key_hash(  select_, key_, hash_ )
		end
	end
	local function func_mysql_()
		if  self.mysqlDcClient then
			self.mysqlDcClient:unlock_key_hash( select_, key_, hash_ )
		end
	end

	if  self.order_type then
		local thread_list_ = {}
		uu.thread_spawn( func_ssdb_,  thread_list_ )
		uu.thread_spawn( func_mysql_, thread_list_ )
		uu.thread_wait( thread_list_ )
	else
		func_ssdb_()
		func_mysql_()
	end
end


------------------------------------
---批量读取 (不会对数据加锁)
function _M.mread( self, select_, kv_list_, err_call_back )
	local ret_ssdb_, ret_mysql_

	local function func_ssdb_()
		if  self.ssdbDcClient then
			ret_ssdb_ = self.ssdbDcClient:mread(   select_, kv_list_, err_call_back )
		end
	end
	local function func_mysql_()
		if  self.mysqlDcClient then
			ret_mysql_ = self.mysqlDcClient:mread( select_, kv_list_, err_call_back )
		end
	end


	if  self.order_type then
		-- 开启了双落地模式
		if  self.if_check_mread_ret == 1 then
			-- 需要读取两种模式来比较
		else
			-- 不做mread比较，只需要读一种
			if  self.order_type == __MIX_ORDER_MYSQL_SSDB_1__ then
				func_mysql_()
				return ret_mysql_
			else
				func_ssdb_()
				return ret_ssdb_
			end
		end


		local thread_list_ = {}
		uu.thread_spawn( func_ssdb_,  thread_list_ )
		uu.thread_spawn( func_mysql_, thread_list_ )
		uu.thread_wait( thread_list_ )

		if  ret_ssdb_ and ret_mysql_ then
			-- 进行值比较
			if  uu.tableNotEmpty( ret_ssdb_ ) then
				local err_cc_= 0
				for i=1, #ret_ssdb_ do
					if  ret_ssdb_[i] == ngx.null or ret_mysql_[i] == ngx.null then
						-- 有一方为null不比较
					else
						local s1_ = uu.to_str(ret_ssdb_[i])
						local s2_ = uu.to_str(ret_mysql_[i])
						if  s1_ == s2_ then
							--match ok
						else
							log.day_list( "mix_dc", "match_error_mread1", select_, #kv_list_, i, "\n", s1_, "\n", s2_ )
							err_cc_ = err_cc_ + 1
						end
					end

				end

				if  err_cc_ > 0 then
					log.day_list( "mix_dc", "match_error_mread2", select_, #kv_list_, #ret_ssdb_, #ret_mysql_, err_cc_)
					--, debug.traceback() )
				else
					if  log.fast() then
						log.debug_list( "mix_dc", "match_mread",  select_, #kv_list_, #ret_ssdb_, #ret_mysql_, err_cc_ )
					end
				end
			end
		end

		if  self.order_type == __MIX_ORDER_MYSQL_SSDB_1__ then
			return ret_mysql_
		else
			return ret_ssdb_
		end
	else
		func_ssdb_()
		func_mysql_()
		return (ret_ssdb_ or ret_mysql_)
	end

end



----主动对数据进行加远程锁 (基本上不需要调用)
-- 一般read操作的时候会加锁， save或者final解锁
-- 特殊情况下需要直接加远程锁，目前只有ma服务在使用
-- lock_id_ = 锁id  mysql模式下可选参数
function _M.add_read_lock( self, select_, key, hash, lock_id_ )
	if  self.ssdbDcClient then
		self.ssdbDcClient:add_read_lock(  select_, key, hash )
	end
	if  self.mysqlDcClient then
		self.mysqlDcClient:add_read_lock( select_, key, hash, lock_id_ )
	end
end



---读取单个数据，可选加远程锁
function _M.read( self, select_, key, hash, err_call_back, no_cache_, read_only_ )
	local ret_ssdb_, ret_mysql_
	local function func_ssdb_()
		if  self.ssdbDcClient then
			ret_ssdb_ = self.ssdbDcClient:read(   select_, key, hash, err_call_back, no_cache_, read_only_ )
		end
	end

	local function func_mysql_()
		if  self.mysqlDcClient then
			ret_mysql_ = self.mysqlDcClient:read( select_, key, hash, err_call_back, no_cache_, read_only_ )
		end
	end

	if  self.order_type then
		local thread_list_ = {}
		uu.thread_spawn( func_ssdb_,  thread_list_ )
		uu.thread_spawn( func_mysql_, thread_list_ )
		uu.thread_wait( thread_list_ )

		if (self.if_check_read_ret == 1) and ret_ssdb_ and ret_mysql_ then
			-- 进行值比较
			if  ret_ssdb_ == ngx.null or ret_mysql_ == ngx.null then
				-- 有一方为null不比较
			else
				local s1_ = uu.to_str(ret_ssdb_)
				local s2_ = uu.to_str(ret_mysql_)
				if  s1_ == s2_ then
					--match ok
				else
					log.day_list( "mix_dc", "not_match", select_, key, hash )
					if  self.repair_data == 1 then
						-- 需要修复数据
						if     self.order_type == __MIX_ORDER_SSDB_MYSQL_0__ then
							--- ssdb 修复同步到 mysql
							if  type(ret_ssdb_) == 'string' and #ret_ssdb_ > 0 then
								self.mysqlDcClient:save( select_, key, hash, ret_ssdb_,  nil, nil, 1 )
								log.day_list( "mix_dc", "repair_data_mysql", select_, key, hash )
							else
								log.day_list( "mix_dc", "error_repair_data_mysql", select_, key, hash, uu.to_str(ret_ssdb_) )
							end

						elseif self.order_type == __MIX_ORDER_MYSQL_SSDB_1__ then
							--- mysql 修复同步到 ssdb
							if  type(ret_mysql_) == 'string' and #ret_mysql_ > 0 then
								self.ssdbDcClient:save(  select_, key, hash, ret_mysql_, nil, nil, 1 )
								log.day_list( "mix_dc", "repair_data_ssdb",  select_, key, hash )
							else
								log.day_list( "mix_dc", "error_repair_data_ssdb", select_, key, hash, uu.to_str(ret_mysql_) )
							end
						end
					else
						log.day_list( "mix_dc", "not_match", select_, key, hash )
					end
				end

			end

		end


		if  self.order_type == __MIX_ORDER_MYSQL_SSDB_1__ then
			return ret_mysql_
		else
			return ret_ssdb_
		end
	else
		func_ssdb_()
		func_mysql_()
		return (ret_ssdb_ or ret_mysql_)
	end

end



---写入单个数据，并解开远程锁
function _M.save( self, select_, key, hash, value, err_call_back, delay_, read_only_, ver_ )
	local ret_ssdb_, ret_mysql_

	local function func_ssdb_()
		if  self.ssdbDcClient then
			ret_ssdb_ = self.ssdbDcClient:save(   select_, key, hash, value, err_call_back, delay_, read_only_ )
		end
	end
	local function func_mysql_()
		if  self.mysqlDcClient then
			ret_mysql_ = self.mysqlDcClient:save( select_, key, hash, value, err_call_back, delay_, read_only_, ver_ )
		end
	end

	if  self.order_type then
		local thread_list_ = {}
		uu.thread_spawn( func_ssdb_,  thread_list_ )
		uu.thread_spawn( func_mysql_, thread_list_ )
		uu.thread_wait( thread_list_ )

		if  self.order_type == __MIX_ORDER_MYSQL_SSDB_1__ then
			return ret_mysql_
		else
			return ret_ssdb_
		end
	else
		func_ssdb_()
		func_mysql_()
		return (ret_ssdb_ or ret_mysql_)
	end

end


--- 删除一个key下的一个hash，也删除对应的value
function _M.del( self, select, key, hash, err_call_back )
	local ret_ssdb_, ret_mysql_

	local function func_ssdb_()
		if  self.ssdbDcClient then
			ret_ssdb_ = self.ssdbDcClient:del(   select, key, hash, err_call_back )
		end
	end
	local function func_mysql_()
		if  self.mysqlDcClient then
			ret_mysql_ = self.mysqlDcClient:del( select, key, hash, err_call_back )
		end
	end

	if  self.order_type then
		local thread_list_ = {}
		uu.thread_spawn( func_ssdb_,  thread_list_ )
		uu.thread_spawn( func_mysql_, thread_list_ )
		uu.thread_wait( thread_list_ )

		if  self.order_type == __MIX_ORDER_MYSQL_SSDB_1__ then
			return ret_mysql_
		else
			return ret_ssdb_
		end

	else
		func_ssdb_()
		func_mysql_()
		return (ret_ssdb_ or ret_mysql_)
	end

end


--- 获得一个key下的所有hash值
function _M.hlist( self, select, start_, end_, size_, err_call_back)
	local ret_ssdb_, ret_mysql_

	local function func_ssdb_()
		if  self.ssdbDcClient then
			ret_ssdb_ = self.ssdbDcClient:hlist(   select, start_, end_, size_, err_call_back)
		end
	end
	local function func_mysql_()
		if  self.mysqlDcClient then
			ret_mysql_ = self.mysqlDcClient:hlist( select, start_, end_, size_, err_call_back)
		end
	end

	if  self.order_type then
		local thread_list_ = {}
		uu.thread_spawn( func_ssdb_,  thread_list_ )
		uu.thread_spawn( func_mysql_, thread_list_ )
		uu.thread_wait( thread_list_ )

		if  self.order_type == __MIX_ORDER_MYSQL_SSDB_1__ then
			return ret_mysql_
		else
			return ret_ssdb_
		end
	else
		func_ssdb_()
		func_mysql_()
		return (ret_ssdb_ or ret_mysql_)
	end

end


--- 获得一个key下有多少key hash组合
function _M.hgetall( self, select, key_, err_call_back)
	local ret_ssdb_, ret_mysql_

	local function func_ssdb_()
		if  self.ssdbDcClient then
			ret_ssdb_ = self.ssdbDcClient:hgetall(   select, key_, err_call_back)
		end
	end
	local function func_mysql_()
		if  self.mysqlDcClient then
			ret_mysql_ = self.mysqlDcClient:hgetall( select, key_, err_call_back)
		end
	end

	if  self.order_type then
		local thread_list_ = {}
		uu.thread_spawn( func_ssdb_,  thread_list_ )
		uu.thread_spawn( func_mysql_, thread_list_ )
		uu.thread_wait( thread_list_ )

		if  self.order_type == __MIX_ORDER_MYSQL_SSDB_1__ then
			return ret_mysql_
		else
			return ret_ssdb_
		end
	else
		func_ssdb_()
		func_mysql_()
		return (ret_ssdb_ or ret_mysql_)
	end

end


--- 一个key下有多少个hash值
function _M.hlen( self, select, key, err_call_back)
	local ret_ssdb_, ret_mysql_
	local function func_ssdb_()
		if  self.ssdbDcClient then
			ret_ssdb_ = self.ssdbDcClient:hlen(   select, key, err_call_back)
		end
	end
	local function func_mysql_()
		if  self.mysqlDcClient then
			ret_mysql_ = self.mysqlDcClient:hlen( select, key, err_call_back)
		end
	end

	if  self.order_type then
		local thread_list_ = {}
		uu.thread_spawn( func_ssdb_,  thread_list_ )
		uu.thread_spawn( func_mysql_, thread_list_ )
		uu.thread_wait( thread_list_ )

		if  self.order_type == __MIX_ORDER_MYSQL_SSDB_1__ then
			return ret_mysql_
		else
			return ret_ssdb_
		end
	else
		func_ssdb_()
		func_mysql_()
		return (ret_ssdb_ or ret_mysql_)
	end

end


---获得一个key下的所有值
-----[key][hash1]=[value1]
-----[key][hash2]=[value2]
-----[key][hash3]=[value3]
function _M.hget_all( self, select, key, err_call_back )
	local ret_ssdb_, ret_mysql_

	local function func_ssdb_()
		if  self.ssdbDcClient then
			ret_ssdb_ = self.ssdbDcClient:hget_all(   select, key, err_call_back )
		end
	end
	local function func_mysql_()
		if  self.mysqlDcClient then
			ret_mysql_ = self.mysqlDcClient:hget_all( select, key, err_call_back )
		end
	end

	if  self.order_type then
		local thread_list_ = {}
		uu.thread_spawn( func_ssdb_,  thread_list_ )
		uu.thread_spawn( func_mysql_, thread_list_ )
		uu.thread_wait( thread_list_ )
		if  self.order_type == __MIX_ORDER_MYSQL_SSDB_1__ then
			return ret_mysql_
		else
			return ret_ssdb_
		end
	else
		func_ssdb_()
		func_mysql_()
		return (ret_ssdb_ or ret_mysql_)
	end

end


---设置单个kv
function _M.set_kv( self, select, key, value, et, err_call_back )
	local ret_ssdb_, ret_mysql_
	if  self.ssdbDcClient then
		ret_ssdb_ = self.ssdbDcClient:set_kv(  select, key, value, et, err_call_back )
	end
	if  self.mysqlDcClient then
		--haima todo
		--ret_mysql_ = self.mysqlDcClient:set_kv( select, key, value, et, err_call_back )
	end
	return (ret_ssdb_ or ret_mysql_)
end


---获得单个kv
function _M.get_kv( self, select, key, err_call_back )
	local ret_ssdb_, ret_mysql_
	if  self.ssdbDcClient then
		ret_ssdb_ = self.ssdbDcClient:get_kv(  select, key, key, err_call_back )
	end
	if  self.mysqlDcClient then
		--haima todo
		--ret_mysql_ = self.mysqlDcClient:get_kv( select, key, key, err_call_back )
	end
	return (ret_ssdb_ or ret_mysql_)
end


---发送远程解锁请求，解锁一个key hash对
---(一般情况下read加锁，save解锁，或者 final() 统一解锁，很少这样去主动解锁)
function _M.send_unlock_kh_to_dc( self, key_hash_, select_, err_call_back, lock_id_ )
	local ret_ssdb_, ret_mysql_

	local function func_ssdb_()
		if  self.ssdbDcClient then
			ret_ssdb_ = self.ssdbDcClient:send_unlock_kh_to_dc(   key_hash_, select_, err_call_back )
		end
	end
	local function func_mysql_()
		if  self.mysqlDcClient then
			ret_mysql_ = self.mysqlDcClient:send_unlock_kh_to_dc( key_hash_, select_, err_call_back, lock_id_ )
		end
	end

	if  self.order_type then
		local thread_list_ = {}
		uu.thread_spawn( func_ssdb_,  thread_list_ )
		uu.thread_spawn( func_mysql_, thread_list_ )
		uu.thread_wait( thread_list_ )

		if  self.order_type == __MIX_ORDER_MYSQL_SSDB_1__ then
			return ret_mysql_
		else
			return ret_ssdb_
		end
	else
		func_ssdb_()
		func_mysql_()
		return (ret_ssdb_ or ret_mysql_)
	end

end


---远程清理缓存， 后续的读请求会重新加载ssdb数据
function _M.clean_shared_cache_ssdb(self, select_, key_, hash_ )
	local ret_ssdb_, ret_mysql_

	local function func_ssdb_()
		if  self.ssdbDcClient then
			ret_ssdb_ = self.ssdbDcClient:clean_shared_cache_ssdb(   select_, key_, hash_ )
		end
	end
	local function func_mysql_()
		if  self.mysqlDcClient then
			ret_mysql_ = self.mysqlDcClient:clean_shared_cache_ssdb( select_, key_, hash_ )
		end
	end

	if  self.order_type then
		local thread_list_ = {}
		uu.thread_spawn( func_ssdb_,  thread_list_ )
		uu.thread_spawn( func_mysql_, thread_list_ )
		uu.thread_wait( thread_list_ )

		if  self.order_type == __MIX_ORDER_MYSQL_SSDB_1__ then
			return ret_mysql_
		else
			return ret_ssdb_
		end
	else
		func_ssdb_()
		func_mysql_()
		return (ret_ssdb_ or ret_mysql_)
	end

end



---------------------------------------------校验函数----------------------------
--写校验[_k_] [_h_]
_M.private_check_set_key = function(self, select_, key, hash, value )
	if  value._k_ then
		if  ( tonumber(value._k_) and ( tonumber(key) == tonumber(value._k_) ) ) or (key == value._k_) then
			--ok
		else
			log.day_list_short( "mix_dc", "error_key_check", "set", select_, key, value._k_, hash, debug.traceback() );
			log.error( "error_set_key_check" )
			return false;
		end
	else
		value._k_ = tonumber(key) or key;  --首次设置
	end

	if  value._h_ then
		if  ( tonumber(value._h_) and ( tonumber(hash) == tonumber(value._h_) ) ) or (hash == value._h_) then
			--ok
		else
			log.day_list_short( "mix_dc", "error_hash_check", "set", select_, key, value._h_, hash );
			log.error( "error_set_hash_check" )
			return false;
		end
	else
		if  tonumber(hash) then
			value._h_ = tonumber(hash) --首次设置
		end
	end

	value._t_ = uu.now()    --最后一次写入时间
	return true;
end



--读校验[_k_] [_h_]
_M.private_check_get_key = function(self, select_, key, hash, value )
	if  value._k_ then
		if  ( tonumber(value._k_) and ( tonumber(key) == tonumber(value._k_) ) ) or (key == value._k_) then
			--ok
		else
			--读取出错
			log.day_list_short( "mix_dc", "error_key_check", "get", select_, key, value._k_, hash, debug.traceback() );
			log.error( "error_get_key_check" )
			return false;
		end
	end

	if  value._h_ then
		if  ( tonumber(value._h_) and ( tonumber(hash) == tonumber(value._h_) ) ) or (hash == value._h_) then
			--ok
		else
			--读取出错
			log.day_list_short( "mix_dc", "error_hash_check", "get", select_, key, value._h_, hash );
			log.error( "error_get_hash_check" )
			return false;
		end
	end

	return true;
end



return _M;
